﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using PServiceBus.Core.Interface;
using Phoenix.ReflectionExtensions;
using PServiceBus.Core.Runtime.Transports;
using PServiceBus.Core.Runtime;
using System.Text.RegularExpressions;
using System.Reflection;
using System.Collections.Concurrent;
using System.Linq.Expressions;
using PServiceBus.Gateway.Interfaces.Exceptions;
using PServiceBus.Gateway.Interfaces;
using PServiceBus.Services.Gateway.Extensions;
using DTO = PServiceBus.Gateway.Interfaces.DTO;
using PServiceBus.Core.Runtime.Extensions;

namespace PServiceBus.Services.Gateway.Runtime {
    /// <summary>
    /// This class is used for subscribing to topics and creating subscribers
    /// </summary>
    public sealed class Subscriber {
        private static Regex _topicPatternRegex = new Regex(@"(\*\.)|(\.\*)|(\*)", RegexOptions.Compiled | RegexOptions.Multiline | RegexOptions.IgnoreCase);
        private Dictionary<string, DTO.TransportInfo> _transports =
            new Dictionary<string, DTO.TransportInfo>();
        private Dictionary<string, Topic> _topics = new Dictionary<string, Topic>();
        internal IList<Action<IMessageBus>> _actions = new List<Action<IMessageBus>>();
        private string _name;
        private Guid _id;
        private bool _durable;

        private Subscriber(string name, bool isNew = true) {
            _name = name;
            if (isNew) _actions.Add(connector => connector.CreateSubscriber(_name));
            IsValid = true;
        }
        /// <summary>
        /// Return an instance of a subscriber with the specified name
        /// </summary>
        /// <param name="name"></param>
        /// <returns></returns>
        public static Subscriber Select(string name) {
            Subscriber subscriber = null;
            ESB.Execute(connector => subscriber = connector.SelectSubscriber(name).ToSubscriber());
            if (subscriber != null)
                subscriber._actions.Clear();
            return subscriber ?? Subscriber.New(name).Valid(false);
        }


        public bool IsValid { get; private set; }
        public Dictionary<string, DTO.TransportInfo> Transports { get { return _transports; } }
        public Dictionary<string, Topic> Topics { get { return _topics; } }
        public Guid SubscriberID { get { return _id; } }
        public string SubscriberName { get { return _name; } }
        public bool IsSubscriberDurable { get { return _durable; } internal set { _durable = value; } }
        /// <summary>
        /// Change the current subscriber's name
        /// </summary>
        /// <param name="name"></param>
        /// <returns></returns>
        public Subscriber Name(string name) {
            ValidateState();
            _name = name;
            return this;
        }
        /// <summary>
        /// Change the current subscriber's id
        /// </summary>
        /// <param name="id"></param>
        /// <returns></returns>
        internal Subscriber Id(Guid id) {
            ValidateState();
            _id = id;
            return this;
        }
        
        /// <summary>
        /// Change the state of the subscriber
        /// </summary>
        /// <param name="value"></param>
        /// <returns></returns>
        internal Subscriber Valid(bool value) {
            IsValid = value;
            return this;
        }

        /// <summary>
        /// Set the durability of the subscriber
        /// </summary>
        /// <param name="durable"></param>
        /// <returns></returns>
        public Subscriber Durable(bool durable) {
            ValidateState();
            _durable = durable;
            _actions.Add(connector => connector.ChangeSubscriberDurability(_name, _durable));
            return this;
        }

        public bool IsSubscribeTo<TTopic>() where TTopic : class {
            var topic = typeof(TTopic).Name;
            return IsSubscribeTo(topic);
        }

        public bool IsSubscribeTo(string topic) {
            return Topics.ContainsKey(topic);
        }

        public bool IsTransportRegistered(string transportName) {
            return Transports.ContainsKey(transportName);
        }
        
        /// <summary>
        /// Add a transport for receiving message to the current subscriber
        /// If topicName is null or not specified, transport will be applied to all topic by default
        /// </summary>
        /// <param name="name"></param>
        /// <param name="transport"></param>
        /// <param name="topicName"></param>
        /// <returns></returns>
        public Subscriber AddTransport(string name, ITransport transport, string topicName = null) {
            ValidateState();
            _transports[name] = new DTO.TransportInfo() { TopicName = topicName, Name = name, Transport = transport };
            _actions.Add(connector => connector.AddTransport(_name, name, transport, topicName));
            return this;
        }

        /// <summary>
        /// Add a transport for receiving message to the current subscriber for specified topic type
        /// </summary>
        /// <typeparam name="TTopic"></typeparam>
        /// <param name="name"></param>
        /// <param name="transport"></param>
        /// <returns></returns>
        public Subscriber AddTransportFor<TTopic>(string name, ITransport transport) {
            var type = typeof(TTopic);
            return AddTransport(name, transport, topicName: type.Name);
        }

        internal static Type _transportHandler = typeof(ITransportHandler<>);

        /// <summary>
        /// Create a message handler based on the transport that is defined the for specified transport name
        /// Will automatically start the handler if everything load successfully
        /// </summary>
        /// <typeparam name="TMessage">Type to convert message payload to</typeparam>
        /// <param name="transportName">Name of transport that will expose the message handler</param>
        /// <param name="action">action to invoke every time a message is received</param>
        /// <param name="errorAction">action to invoke when exception occurs</param>
        /// <param name="interval">Default is 100 millisecond</param>
        /// <param name="batchSize">Default is 1</param>
        public IMessageHandler<TMessage> OnMessageReceived<TMessage>(string transportName, Action<TMessage> action = null,
            Action<HandleErrorArgs<TMessage>> errorAction= null, TimeSpan interval = default(TimeSpan), int batchSize = 1,
            bool handleInListenerThread = true, Action<IList<TMessage>> actionList = null) where TMessage : class {
            if (!_transports.ContainsKey(transportName)) 
                throw new ESBException(String.Format("The specified transport[{0}] does not exist for the current subscriber", transportName));
            var transportInfo = _transports[transportName];
            var transport = transportInfo.Transport;
            var messageHandler = null as IMessageHandler<TMessage>;
            var transportHandlerName = _transportHandler.MakeGenericType(transport.GetType()).FullName;
            var asmTypes = AppDomain.CurrentDomain.GetAssemblies().SelectMany(a => a.GetTypes());
            var transportHandlerType =
                asmTypes.FirstOrDefault(t => t.GetInterfaces().Any(ti => ti.FullName == transportHandlerName));
            if (transportHandlerType == null)
                throw new ESBException(String.Format("{0} does not have any transport handler to create a message handler for the specific transport[{1}]",
                    transport.GetType().Name, transportName));
            var transportHandler = (transportHandlerType != null ? Activator.CreateInstance(transportHandlerType) : null);
            if (transportHandler != null) {
                var type = transportHandler.GetType();
                var createHandlerMethod = type.GetMethodEx("CreateHandler");
                var genericMethod = createHandlerMethod.MakeGenericMethod(typeof(TMessage));
                type.GetPropertyEx("Interval").SetValueEx(transportHandler, interval == default(TimeSpan) ? TimeSpan.FromMilliseconds(100) : interval);
                type.GetPropertyEx("BatchSize").SetValueEx(transportHandler, batchSize);
                messageHandler = genericMethod.Invoke(transportHandler, new object[] { transport }) as IMessageHandler<TMessage>;
            }
            if (messageHandler != null) {
                messageHandler.HandleInListenerThread = handleInListenerThread;
                messageHandler.MessageReceived += (s, e) => {
                    if (action != null) {
                        foreach (var msg in e.Messages)
                            action(msg);
                    }
                    if (actionList != null) 
                        actionList(e.Messages);
                };
                messageHandler.ErrorReceived += (s, e) =>
                {
                    if (errorAction != null) 
                        errorAction(e);
                };
                messageHandler.Start();
                TransportHandlers.AddHandler(messageHandler);
            }
            return messageHandler;
        }

        /// <summary>
        /// Delete a transport if exist from the current subscriber
        /// Will throw exception if transport does not exists
        /// </summary>
        /// <param name="name"></param>
        /// <returns></returns>
        public Subscriber DeleteTransport(string name) {
            ValidateState();
            _transports.Remove(name);
            _actions.Add(connector => connector.DeleteTransport(_name, name));
            return this;
        }
        /// <summary>
        /// Create a new subscriber instance based on the specified name
        /// This is used to create subscriber determine if they are newly created
        /// or if the instance is used to regenerate an existing subscriber
        /// </summary>
        /// <param name="name"></param>
        /// <param name="isNew">if false, it indicate that it is not creating a new subscriber</param>
        /// <returns></returns>
        internal static Subscriber New(string name, bool isNew = true) {
            return new Subscriber(name, isNew);
        }
        /// <summary>
        /// Create a new subscriber instance based on the specified name
        /// </summary>
        /// <param name="name"></param>
        /// <returns></returns>
        public static Subscriber New(string name) {
            return new Subscriber(name, isNew: true);
        }

        /// <summary>
        /// Return subscriber if exists else return new subscriber based on name
        /// </summary>
        /// <param name="name"></param>
        /// <returns></returns>
        public static Subscriber SelectOrAdd(string name) {
            var subscriber = Subscriber.Select(name);
            if (!subscriber.IsValid) Subscriber.New(name).Save();
            return Subscriber.Select(name);
        }

        /// <summary>
        /// Update existing subscription for a topic.
        /// This operation is useful when there is a need to refresh filters and case sensitive of message filters
        /// It removes the need to unsubscribe in order to resubscribe
        /// Will throw an exception if subscription does not exists already for topic
        /// </summary>
        /// <param name="topic"></param>
        /// <returns></returns>
        public Subscriber Update(Topic topic) {
            ValidateState();
            if (!topic.IsValid) throw new TopicNotRegisteredException(ExceptionConstants.TopicExceptionStr(topic.TopicName));
            if (!_topics.ContainsKey(topic.TopicName))
                throw new InvalidOperationException("Subscriber must be subscribed to topic in order to perform this operation: Update");

            _topics[topic.TopicName] = topic;
            _actions.Add(connector => connector.UpdateSubscriber(_name,
                topic.TopicName, topic.TopicFilter, topic.NeedMessageHeaders, topic.CaseSensitive));
            
            return this;
        }

        /// <summary>
        /// Subscribe to the specified topic
        /// </summary>
        /// <param name="topic"></param>
        /// <returns></returns>
        public Subscriber SubscribeTo(Topic topic) {
            ValidateState();
            if (_topicPatternRegex.IsMatch(topic.TopicName)) {
                if (!topic.IsValid) {
                    var filters = topic.TopicFilter;
                    topic = Topic.New(topic.TopicName).Filter(filters).Valid(true);
                    var ex = MethodHelper.Try(() => topic.Register());
                    if (ex != null && ex is TopicAlreadyExistException ) 
                        topic = Topic.Select(topic.TopicName);
                }
            }
            if (!topic.IsValid) throw new TopicNotRegisteredException(ExceptionConstants.TopicExceptionStr(topic.TopicName));
            _topics[topic.TopicName] = topic;
            _actions.Add(connector => connector.SubscribeTo(_name, topic.TopicName, topic.TopicFilter, topic.NeedMessageHeaders, topic.CaseSensitive));
            return this;
        }
        /// <summary>
        /// Un-subscribe from the specified topic name
        /// </summary>
        /// <param name="topicName"></param>
        /// <returns></returns>
        public Subscriber UnSubscribeFrom(string topicName) {
            ValidateState();
            _topics.Remove(topicName);
            _actions.Add(connector => connector.UnSubscribeFrom(_name, topicName));
            return this;
        }
        /// <summary>
        /// Delete the current subscriber
        /// </summary>
        public void Delete() {
            ValidateState();
            ESB.Execute(connector => connector.DeleteSubscriber(_name));
        }
        /// <summary>
        /// Commit the changes that have be done to the current instance of the subscriber
        /// Save has to be called after anything changes with the subscriber
        /// </summary>
        public void Save() {
            ValidateState();
            ESB.Execute(connector => {
                foreach (var action in _actions) action(connector);
            });
            _actions.Clear();
        }
        /// <summary>
        /// Throw an exception if subscriber is currently in an invalid state
        /// </summary>
        private void ValidateState() {
            if (!IsValid) throw new SubscriberNotExistException(ExceptionConstants.SubscriberExceptionStr(_name));
        }
    }
    public static partial class SubscriberExtension {
        private static readonly MethodInfo MessageReceivedMethod = typeof(Subscriber).GetMethod("OnMessageReceived");


        private static readonly ConcurrentDictionary<string, object> _cacheListActions =
            new ConcurrentDictionary<string, object>();
        private static readonly Type _ilistType = typeof(IList<>);
        private static readonly MethodInfo _toObjects =
            typeof(SubscriberExtension).GetMethod("ToObjects", BindingFlags.Static | BindingFlags.NonPublic);

        private static List<object> ToObjects<T>(IList<T> list) {
            return list.Cast<object>().ToList();
        }

        private static object GetListAction(Type type, Action<IList<object>> listAction) {
            object action = null;
            var key = type.FullName;
            if (listAction == null || _cacheListActions.TryGetValue(key, out action)) return action;
            var listType = _ilistType.MakeGenericType(type);
            var toObjects = _toObjects.MakeGenericMethod(type);
            var p = Expression.Parameter(listType, "msg");
            action =
                Expression.Lambda(Expression.GetActionType(listType),
                Expression.Call(
                    listAction.Target == null ? null : Expression.Constant(listAction.Target),
                    listAction.Method, Expression.Call(null, toObjects, p)), p).Compile();
            return _cacheListActions[key] = action;
        }

        public static object OnMessageReceived(this Subscriber subscriber, string transportName, Dictionary<string, object> structure,
            Action<object> action = null, Action<HandleErrorArgs<object>> errorAction = null, TimeSpan interval = default(TimeSpan), int batchSize = 1,
            bool handleInListenerThread = true, Action<IList<object>> actionList = null) {
                var className = structure.GenerateClassName();
            var messageType = ReflectionHelper.GenerateClassType(className, structure);
            var genericMethod = MessageReceivedMethod.MakeGenericMethod(messageType);
            errorAction = errorAction ?? new Action<HandleErrorArgs<object>>(errorHandler => { errorHandler.Continue = true; });
            interval = interval == default(TimeSpan) ? TimeSpan.FromMilliseconds(100) : interval;
            var handler = genericMethod.Invoke(subscriber, new object[] { transportName, action, errorAction, interval, batchSize, 
                handleInListenerThread, GetListAction(messageType, actionList)});
            return handler;
        }
    }
}
